/**
 *  julive
 *
 *  Create by songli on 2021/05/24
 *  Copyright © 2021 居理买房网. All rights reserved.
 */

package service

import (
	"context"
	"fmt"
	"io"

	"git.julive.com/julive/user/server/domain"
	"git.julive.com/julive/user/server/proto/user"
	"git.julive.com/micro/pkg/broker"
	"git.julive.com/micro/pkg/logx"
	"git.julive.com/micro/pkg/proto/pb"
	"go.uber.org/zap"
)

type UserService struct {
	user.UnimplementedUserServiceServer

	domain domain.UserDomain

	logger *zap.Logger
}

func NewUserService() *UserService {
	logger := logx.Logger.With(zap.String("module", "comment/service"), zap.String("svc", "comment"))

	dm := domain.NewUserDomain()
	svc := &UserService{
		domain: dm,
		logger: logger,
	}

	return svc
}

// CreateUser 创建User
func (svc *UserService) CreateUser(ctx context.Context, req *user.CreateUserRequest) (*user.CreateUserResponse, error) {
	rsp := &user.CreateUserResponse{}

	m := req.Dto
	if m == nil {
		err := &pb.Error{}
		err.Message = "User不能为空"
		rsp.Error = err
		return rsp, nil
	}

	if len(m.Name) == 0 {
		err := &pb.Error{}
		err.Message = "Name不能为空"
		rsp.Error = err
		return rsp, nil
	}

	err := svc.domain.CreateUser(ctx, m)
	if err != nil {
		err := &pb.Error{}
		err.Message = "创建User失败"
		rsp.Error = err
		return rsp, nil
	}

	rsp.Success = true

	endpoint, err := broker.NewBroker("endpoint")
	if err != nil {
		svc.logger.Error("broker.NewBroker error", zap.Error(err))
		return rsp, err
	}
	endpoint.Publish(&broker.Message{
		Topic: endpoint.Options().Topics["createuser"],
		Body:  []byte(fmt.Sprintf("User %s Created", m.Name)),
	})

	return rsp, nil
}

// GetUserById 根据Id获取User
func (svc *UserService) GetUserById(ctx context.Context, req *user.GetUserByIdRequest) (*user.GetUserByIdResponse, error) {
	fmt.Println("User Received UserService.GetUserById")
	rsp := &user.GetUserByIdResponse{}
	m, err := svc.domain.GetUserById(ctx, req.Id)
	if err != nil {
		err := &pb.Error{}
		err.Message = "获取User失败"
		rsp.Error = err
		return rsp, nil
	}
	rsp.Success = true
	rsp.Dto = m
	return rsp, nil
}

func (svc *UserService) GetUserList(ctx context.Context, req *user.GetUserListRequest) (*user.GetUserListResponse, error) {
	rsp := &user.GetUserListResponse{}
	list, err := svc.domain.GetUserList(ctx, req.Status, req.Lastid, req.Pagesize, req.Page)
	if err != nil {
		err := &pb.Error{}
		err.Message = "获取UserList失败"
		rsp.Error = err
		return rsp, nil
	}
	rsp.Success = true
	rsp.Data = list
	return rsp, nil
}

func (svc *UserService) UpdateUserStatus(ctx context.Context, req *user.UpdateUserStatusRequest) (*user.UpdateUserStatusResponse, error) {
	rsp := &user.UpdateUserStatusResponse{}
	affected, err := svc.domain.UpdateUserStatus(ctx, req.Id, req.Status)
	if err != nil || affected <= 0 {
		err := &pb.Error{}
		err.Message = "更新User失败"
		rsp.Error = err
		return rsp, nil
	}
	rsp.Success = true
	return rsp, nil
}

func (svc *UserService) UpdateUserCount(ctx context.Context, req *user.UpdateUserCountRequest) (*user.UpdateUserCountResponse, error) {
	rsp := &user.UpdateUserCountResponse{}
	affected, err := svc.domain.UpdateUserCount(ctx, req.Id, req.Num, req.Column)
	if err != nil || affected <= 0 {
		err := &pb.Error{}
		err.Message = "更新User失败"
		rsp.Error = err
		return rsp, nil
	}
	rsp.Success = true

	endpoint, err := broker.NewBroker("endpoint")
	if err != nil {
		svc.logger.Error("kafka.New error", zap.Error(err))
		return rsp, err
	}
	endpoint.Publish(&broker.Message{
		Topic: endpoint.Options().Topics["updateusercount"],
		Body:  nil,
	})

	return rsp, nil
}

func (svc *UserService) DeleteUserById(ctx context.Context, req *user.DeleteUserByIdRequest) (*user.DeleteUserByIdResponse, error) {
	rsp := &user.DeleteUserByIdResponse{}
	affected, err := svc.domain.DeleteUserById(ctx, req.Id)
	if err != nil || affected <= 0 {
		err := &pb.Error{}
		err.Message = "删除User失败"
		rsp.Error = err
		return rsp, nil
	}
	rsp.Success = true
	return rsp, nil
}

func (svc *UserService) GetUserListByStream(stream user.UserService_GetUserListByStreamServer) error {
	logger := svc.logger.With(zap.String("func", "GetUserListByStream"))
	for {
		r, err := stream.Recv()
		logger.Debug("stream.Recv", zap.Any("r", r), zap.Error(err))
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}

		if r.Id > 0 {
			m, err := svc.domain.GetUserById(stream.Context(), r.Id)
			if err != nil {
				return err
			}

			err = stream.Send(&user.GetUserListByStreamResponse{
				Index:  r.Index,
				Result: m,
			})
			if err != nil {
				return err
			}
		} else {
			_ = stream.Send(&user.GetUserListByStreamResponse{
				Index:  r.Index,
				Result: nil,
			})
		}

	}
}
